package com.roc.rocketmqconsumer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 重试消息-消费者
 *
 * @author Roker
 * @date 2025/07/08
 */
public class RetryMsgConsumer {
    public static void main(String[] args) throws Exception {
        // 1、实例化消费者，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("roc_consumer_group");
        // 2、设置NameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 3、订阅一个或多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TOPIC_RETRY", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 4、设置消息的最大重试次数，超过后进入死信队列
        consumer.setMaxReconsumeTimes(3);
        // 5、注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try{
                    MessageExt msg = msgs.get(0);
                    String msgbody = new String(msg.getBody(), "utf-8");
                    System.out.println(msgbody + "接受到消息: " + msgs);
                    if (msgbody.equals("发送重试消息-4")) {
                        System.err.println("======模拟消费失败-发生错误=======");
                        int a = 1 / 0; // 模拟消费异常
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    // 如果消息重试次数达到最大次数，可进行日志记录等额外处理，这里标记为消费成功让其进入死信队列
                    if (msgs.get(0).getReconsumeTimes() == 3) {
                        System.err.println("======消息消费失败，进入死信队列=======");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }else {
                        // 未达到最大重试次数，返回重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("重试消息消费者启动……");
    }
}
